package com.chiaching.framwork.core.handler.mq;

import com.chiaching.framwork.domian.constant.Action;
import com.chiaching.framwork.domian.constant.RocketTopic;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author jiajing.he
 * @date 2021/7/19 上午10:37
 */
public class MessageListenerImpl implements MessageListenerConcurrently {

    /**
     * k -> topic
     */
    private final ConcurrentHashMap<String, DefaultMessageListener> listenerTables ;

    public MessageListenerImpl(ConcurrentHashMap<String, DefaultMessageListener> listenerTables) {
        this.listenerTables = listenerTables;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            DefaultMessageListener messageListener = listenerTables.get(msg.getTopic());
            Action res = messageListener.consume(msg);
            if (res.equals(Action.RECONSUME_LATER)){
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
